本文档详细分析了Spring Batch项目中Job停止功能的核心实现机制,包括时序图和关键代码分析。
@PostMapping("/api/jobs/{executionId}/stop")
public ResponseEntity<Map<String, Object>> stopJob(@PathVariable Long executionId) {
boolean stopped = jobService.stopJob(executionId);
// 返回停止结果
}
public boolean stopJob(Long executionId) {
// 1. 检查Job是否存在
JobExecution jobExecution = jobExplorer.getJobExecution(executionId);
// 2. 检查Job是否正在运行
if (!jobExecution.getStatus().isRunning()) {
return true;
}
// 3. 设置停止标志
jobStopManager.setStopFlag(executionId);
// 4. 调用Spring Batch停止方法
boolean stopped = jobOperator.stop(executionId);
return true; // 总是返回true,因为停止标志已设置
}
@Component
public class JobStopManager {
// 停止标志存储
private final ConcurrentHashMap<Long, AtomicBoolean> stopFlags = new ConcurrentHashMap<>();
// 线程引用存储
private final ConcurrentHashMap<Long, Thread> jobThreads = new ConcurrentHashMap<>();
public void setStopFlag(Long executionId) {
stopFlags.put(executionId, new AtomicBoolean(true));
// 强制中断对应线程
Thread jobThread = jobThreads.get(executionId);
if (jobThread != null && jobThread.isAlive()) {
jobThread.interrupt();
}
}
public boolean shouldStop(Long executionId) {
AtomicBoolean stopFlag = stopFlags.get(executionId);
return stopFlag != null && stopFlag.get();
}
}
@Async("taskExecutor")
public CompletableFuture<Void> executeBusinessLogicAsync(String batchId, Long batchExecutionId, Long jobExecutionId) {
try {
// 注册当前线程
jobStopManager.registerJobThread(jobExecutionId, Thread.currentThread());
// 业务处理循环
for (Task task : unprocessedTasks) {
// 检查停止标志
if (jobStopManager.shouldStop(jobExecutionId)) {
logger.info("検出到停止信号,退出処理ループ");
break;
}
// 模拟长时间处理
Thread.sleep(batchProperties.getSimulationDurationSeconds() * 1000);
// 处理任务...
}
} finally {
// 清理停止标志和线程引用
jobStopManager.clearStopFlag(jobExecutionId);
}
}
JobOperator.stop()JobStopManager管理停止状态ConcurrentHashMap存储停止标志AtomicBoolean确保标志操作的原子性shouldStop()该停止机制设计体现了企业级应用的高标准要求,通过多层次的停止策略、线程安全的状态管理和完善的资源清理机制,确保了批处理系统的可靠性和稳定性。这种设计模式可以作为其他类似系统的参考实现。